package org.ghost.hbase.util;

import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.io.compress.Compression;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CollectionUtils;
import org.ghost.hbase.model.HbasePutDTO;
import org.ghost.hbase.model.HbasePutsDTO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

public class HbaseUtil implements Closeable {
    private final Logger logger = LoggerFactory.getLogger(HbaseUtil.class);

    private static final int DEFAULT_MAX_VERSIONS = 3;
    private Configuration conf = null;
    private Admin admin = null;
    private Connection conn = null;

    @Override
    public void close() throws IOException {
        if (admin != null) {
            admin.close();
            admin = null;
        }
        if (conn != null) {
            conn.close();
            conn = null;
        }
    }

    /**
     * 初始化连接
     *
     * @throws IOException
     */
    public void init() throws IOException {
        System.setProperty("hadoop.home.dir", "C:/hadoop-2.7.5");
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "HLLOMQ02");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        //conf.set("hbase.master", "192.168.1.100:600000");
        conn = ConnectionFactory.createConnection(conf);
        admin = conn.getAdmin();
    }

    public boolean getHbaseAvailable() {
        try {
            HBaseAdmin.checkHBaseAvailable(conf);
        } catch (ZooKeeperConnectionException | MasterNotRunningException e) {
            logger.error("ZooKeeperConnectionException or MasterNotRunningException", e);
            return false;
        } catch (Exception e) {
            logger.error("Check HBase available throws an Exception. We don't know whether HBase is running or not.", e);
            return false;
        }

        return true;
    }

    public Connection getConnection() {
        if (conn == null) {
            try {
                conn = ConnectionFactory.createConnection(conf);
            } catch (IOException e) {
                logger.error("Create HBase connect error.", e);
            }
        }

        return conn;
    }

    /**
     * 获取HAdmin对象，建表等操作
     *
     * @return
     * @throws IOException
     * @author
     */
    public Admin getHBaseAdmin() throws IOException {
        if (admin == null) {
            admin = getConnection().getAdmin();
        }
        return admin;
    }

    /**
     * 从连接池获取HTable对象
     *
     * @param tableName
     * @return
     * @throws IOException
     * @author
     */
    public Table getTable(String tableName) throws IOException {
        return getConnection().getTable(TableName.valueOf(tableName));
    }

    /**
     * 关闭HTable对象
     *
     * @param table
     * @author
     */
    public void closeTable(Table table) {
        if (table == null) {
            return;
        }
        try {
            table.close();
        } catch (IOException e) {
            logger.error("closeTable error.", e);
        }
    }

    /**
     * 创建表操作
     *
     * @param tableName
     * @param families
     * @author
     */
    public void createTable(String tableName, String[] families) {
        createTable(tableName, DEFAULT_MAX_VERSIONS, null, families);
    }

    /**
     * 创建表操作
     *
     * @param tableName
     * @param splitKeys
     * @param families
     * @author
     */
    public void createTable(String tableName, byte[][] splitKeys, String[] families) {
        createTable(tableName, DEFAULT_MAX_VERSIONS, splitKeys, families);
    }

    /**
     * 创建表操作
     *
     * @param tableName
     * @param maxVersions
     * @param families
     * @author
     */
    public void createTable(String tableName, int maxVersions, String[] families) {
        createTable(tableName, maxVersions, null, families);
    }

    /**
     * 创建表操作
     *
     * @param tableName
     * @param maxVersions
     * @param splitKeys
     * @param families
     */
    public void createTable(String tableName, int maxVersions, byte[][] splitKeys, String[] families) {
        // 参数判空
        if (StringUtils.isBlank(tableName) || families == null || families.length <= 0) {
            return;
        }
        try {
            // 表不存在则创建
            if (!getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
                for (String family : families) {
                    HColumnDescriptor columnDescriptor = new HColumnDescriptor(family);
                    //TODO 需要手动导入libsnappy.so
                    //columnDescriptor.setCompressionType(Compression.Algorithm.SNAPPY);
                    columnDescriptor.setCompressionType(Compression.Algorithm.NONE);
                    columnDescriptor.setMaxVersions(maxVersions);
                    desc.addFamily(columnDescriptor);
                }
                if (splitKeys != null) {
                    getHBaseAdmin().createTable(desc, splitKeys);
                } else {
                    getHBaseAdmin().createTable(desc);
                }
            } else {
                logger.warn("Table " + tableName + " already exists.");
            }
        } catch (IOException e) {
            logger.error("createTable error", e);
        }
    }

    /**
     * 删除指定表名
     *
     * @param tableName
     */
    public void deleteTable(String tableName) {
        try {
            //在删除一张表前，必须先使其失效
            Admin admin = getHBaseAdmin();
            if (admin.tableExists(TableName.valueOf(tableName))) {
                admin.disableTable(TableName.valueOf(tableName));
                admin.deleteTable(TableName.valueOf(tableName));
            }
        } catch (IOException e) {
            logger.error("*****deleteTable出现错误:" + e.getMessage(), e);
        }
    }

    /**
     * 删除指定表名的列簇
     *
     * @param tableName
     * @param family
     */
    public void deleteFamily(String tableName, String family) {
        try {
            admin.deleteColumn(TableName.valueOf(tableName), Bytes.toBytes(family));
        } catch (IOException e) {
            logger.error("*****deleteFamily出现错误:" + e.getMessage(), e);
        }
    }

    /**
     * 查询所有表名
     *
     * @return
     * @throws Exception
     */
    public List<String> getAllTable() throws Exception {
        ArrayList<String> tables = new ArrayList<String>();
        HTableDescriptor[] listTables = admin.listTables();
        if (ArrayUtils.isNotEmpty(listTables)) {
            for (HTableDescriptor tableDesc : listTables) {
                tables.add(tableDesc.getNameAsString());
            }
        }

        return tables;
    }

    /**
     * 查询指定表结构
     *
     * @return
     * @throws Exception
     */
    public HTableDescriptor getTableDesc(String tableName) throws Exception {
        return admin.getTableDescriptor(TableName.valueOf(tableName));
    }

    /**
     * 查询指定表结构
     *
     * @param tableNames
     * @return
     * @throws Exception
     */
    public HTableDescriptor[] getTableDesc(List<String> tableNames) throws Exception {
        return admin.getTableDescriptors(tableNames);
    }

    /**
     * 新增列簇
     *
     * @param tableName
     * @param family
     */
    public void addFamily(String tableName, String family) throws IOException {
        HColumnDescriptor columnDesc = new HColumnDescriptor(family);
        admin.addColumn(TableName.valueOf(tableName), columnDesc);
    }


    /**
     * 查询表信息
     *
     * @param tableName
     */
    public List<Result> getScanData(String tableName) {
        Table hTable = null;
        ResultScanner resultScanner = null;
        List<Result> resultList = new ArrayList<Result>();
        try {
            hTable = getTable(tableName);
            resultScanner = hTable.getScanner(new Scan());
            for (Result rs : resultScanner) {
                resultList.add(rs);
            }
        } catch (IOException e) {
            logger.error("*****getScanData出现错误:" + e.getMessage(), e);
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
            closeTable(hTable);
        }

        return resultList;
    }

    /**
     * scan过滤器
     *
     * @param tableName
     * @param family
     * @param qualifier
     * @throws Exception
     */
    public List<Result> getScanData(String tableName, String family, String qualifier) {
        Table hTable = null;
        ResultScanner resultScanner = null;
        List<Result> resultList = new ArrayList<Result>();
        try {
            hTable = getTable(tableName);
            Scan scan = new Scan();
            scan.addColumn(family.getBytes(), qualifier.getBytes());

            resultScanner = hTable.getScanner(scan);
            for (Result rs : resultScanner) {
                resultList.add(rs);
            }
        } catch (IOException e) {
            logger.error("*****getScanData出现错误:" + e.getMessage(), e);
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
            closeTable(hTable);
        }

        return resultList;
    }

    /**
     * 根据rowKey查询单行
     *
     * @param tableName
     * @param rowKey
     */
    public Result getRecordByRowKey(String tableName, String rowKey) {
        Table hTable = null;
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                return hTable.get(get);
            } else {
                logger.warn("Table " + tableName + " does not exist.");
                return null;
            }
        } catch (IOException e) {
            logger.error("get table [{}] recodes error", tableName, e);
            return null;
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 根据rowKey查询单行
     *
     * @param tableName
     * @param rowKey
     * @param filter    ColumnPaginationFilter filter = new ColumnPaginationFilter(10, 40)
     * @return
     */
    public Result getRecordByRowKey(String tableName, String rowKey, ColumnPaginationFilter filter) {
        Table hTable = null;
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.setFilter(filter);
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                return hTable.get(get);
            } else {
                logger.warn("Table " + tableName + " does not exist.");
                return null;
            }
        } catch (IOException e) {
            logger.error("get table [{}] recodes error", tableName, e);
            return null;
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 根据rowKey查询单行
     *
     * @param tableName
     * @param rowKey
     * @param limit
     * @param offset
     * @return
     */
    public Result getRecordByRowKey(final String tableName, final String rowKey, final int limit, final int offset) {
        Table hTable = null;
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.setFilter(new ColumnPaginationFilter(limit, offset));
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                return hTable.get(get);
            } else {
                logger.warn("Table " + tableName + " does not exist.");
                return null;
            }
        } catch (IOException e) {
            logger.error("get table [{}] recodes error", tableName, e);
            return null;
        } finally {
            closeTable(hTable);
        }
    }


    /**
     * 根据rowKey查询数据
     *
     * @param tableName
     * @param rowKeys
     * @return
     */
    public Result[] getRecordByRowKey(String tableName, List<String> rowKeys) {
        if (CollectionUtils.isEmpty(rowKeys)) {
            logger.warn("Has no rowKeys to get.");
            return null;
        }

        Table hTable = null;
        try {
            List<Get> gets = new ArrayList<>();
            for (String rowKey : rowKeys) {
                gets.add(new Get(Bytes.toBytes(rowKey)));
            }
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                return hTable.get(gets);
            } else {
                logger.warn("Table " + tableName + " does not exist.");
            }
        } catch (IOException e) {
            logger.error("get table [{}] recodes error", tableName, e);
        } finally {
            closeTable(hTable);
        }

        return null;
    }

    /**
     * 获取记录通过Get
     *
     * @param tableName
     * @param get
     * @return
     */
    public Result getRecordByGet(String tableName, Get get) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                return hTable.get(get);
            } else {
                logger.warn("Table " + tableName + " does not exist.");
            }
        } catch (IOException e) {
            logger.error("get table [{}] recodes error", tableName, e);
        } finally {
            closeTable(hTable);
        }
        return null;
    }

    /**
     * 获取记录通过Get
     *
     * @param tableName
     * @param gets
     * @return
     */
    public Result[] getRecordByGet(String tableName, List<Get> gets) {
        if (CollectionUtils.isEmpty(gets)) {
            logger.warn("Has no gets to get.");
            return null;
        }

        Table hTable = null;
        try {
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                return hTable.get(gets);
            } else {
                logger.warn("Table " + tableName + " does not exist.");
            }
        } catch (IOException e) {
            logger.error("get table [{}] recodes error", tableName, e);
        } finally {
            closeTable(hTable);
        }

        return null;
    }

    /**
     * rowFilter使用
     *
     * @param tableName
     * @param reg
     * @throws Exception
     */
    public List<Result> getRowFilter(String tableName, String reg) throws Exception {
        Table hTable = null;
        ResultScanner resultScanner = null;
        List<Result> resultList = new ArrayList<Result>();
        try {
            hTable = getTable(tableName);
            RowFilter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL, new RegexStringComparator(reg));
            Scan s = new Scan();
            s.setFilter(rowFilter);
            resultScanner = hTable.getScanner(s);
            for (Result rs : resultScanner) {
                resultList.add(rs);
            }

        } catch (Exception e) {
            logger.error("*****getRowFilter出现错误:" + e.getMessage(), e);
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
            closeTable(hTable);
        }

        return resultList;
    }

    /**
     * 按过滤条件查询
     *
     * @param tableName
     * @param filter    Filter filter = new SingleColumnValueFilter(Bytes.toBytes("time"), null, CompareFilter.CompareOp.GREATER, Bytes.toBytes("1519953993579"));
     *                  Filter filterList = new FilterList(FilterList.Operator.MUST_PASS_ALL, filter1, filter2, filter3, filter0);
     * @return
     */
    public List<Result> queryByCondition(String tableName, Filter filter) {
        Table hTable = null;
        ResultScanner resultScanner = null;
        List<Result> resultList = new ArrayList<Result>();
        try {
            hTable = getTable(tableName);
            Scan scan = new Scan();
            scan.setCaching(1000);
            scan.setCacheBlocks(true);
            scan.setFilter(filter);
            resultScanner = hTable.getScanner(scan);
            for (Result rs : resultScanner) {
                resultList.add(rs);
            }
        } catch (Exception e) {
            logger.error("*****queryByCondition出现错误:" + e.getMessage(), e);
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
            closeTable(hTable);
        }
        return resultList;
    }

    /**
     * 按过滤条件查询-分页
     *
     * @param tableName
     * @param filter
     * @param limit
     * @param offset
     * @return
     */
    public List<Result> queryByCondition(String tableName, Filter filter, int limit, int offset) {
        Table hTable = null;
        ResultScanner resultScanner = null;
        List<Result> resultList = new ArrayList<Result>();
        try {
            hTable = getTable(tableName);
            Scan scan = new Scan();
            scan.setCaching(1000);
            scan.setCacheBlocks(true);
            scan.setFilter(filter);

            resultScanner = hTable.getScanner(scan);
            int i = 0;
            int begin = offset + 1;
            int end = offset + limit;
            for (Result rs : resultScanner.next(limit + offset)) {
                i++;
                if (i < begin) {
                    //
                } else if (i >= begin && i <= end) {
                    resultList.add(rs);
                } else if (i > end) {
                    break;
                }
            }
        } catch (Exception e) {
            logger.error("*****queryByCondition出现错误:" + e.getMessage(), e);
        } finally {
            if (resultScanner != null) {
                resultScanner.close();
            }
            closeTable(hTable);
        }
        return resultList;
    }

    /**
     * 按过滤条件查询-分页
     *
     * @param tableName
     * @param filter
     * @param startKeyRow
     * @param limit
     * @param offset
     * @return
     */
    public List<Result> queryByCondition(String tableName, Filter filter, String startKeyRow, int limit, int offset) {
        if (StringUtils.isNotBlank(startKeyRow)) {
            Table hTable = null;
            ResultScanner resultScanner = null;
            List<Result> resultList = new ArrayList<Result>();
            try {
                hTable = getTable(tableName);
                Scan scan = new Scan();
                scan.setCaching(1000);
                scan.setCacheBlocks(true);
                scan.setFilter(filter);
                scan.setStartRow(Bytes.toBytes(startKeyRow));
                resultScanner = hTable.getScanner(scan);
                for (Result rs : resultScanner.next(limit + 1)) {
                    if (!Objects.equals(startKeyRow, Bytes.toString(rs.getRow()))) {
                        resultList.add(rs);
                    }
                }
            } catch (Exception e) {
                logger.error("*****queryByCondition出现错误:" + e.getMessage(), e);
            } finally {
                if (resultScanner != null) {
                    resultScanner.close();
                }
                closeTable(hTable);
            }
            return resultList;
        } else {
            return queryByCondition(tableName, filter, limit, offset);
        }
    }

    /**
     * 获取单个列值
     *
     * @param tableName
     * @param rowKey
     * @return
     * @author
     */
    public byte[] get(String tableName, String rowKey, String family, String qualifier) {
        Table hTable = null;
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                Result result = hTable.get(get);
                return result.getValue(Bytes.toBytes(family), Bytes.toBytes(qualifier));
            } else {
                logger.warn("Table " + tableName + " does not exist.");
            }
        } catch (IOException e) {
            logger.error("获取列值失败！ " + e);
        } finally {
            closeTable(hTable);
        }
        return null;
    }

    /**
     * 获取单个列值，字符串返回
     *
     * @param tableName
     * @param rowKey
     * @return
     * @author zhanglei11
     */
    public String getString(String tableName, String rowKey, String family, String qualifier) {
        return Bytes.toString(get(tableName, rowKey, family, qualifier));
    }

    /**
     * 获取一行中某列族的值
     *
     * @param tableName
     * @param rowKey
     * @return
     * @author
     */
    public Map<String, byte[]> getMapByKeyAndFamily(String tableName, String rowKey, String family) {
        Table hTable = null;
        Map<String, byte[]> map = null;
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            get.addFamily(Bytes.toBytes(family));
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                map = new HashMap<String, byte[]>();
                Result result = hTable.get(get);
                for (Cell cell : result.rawCells()) {
                    byte[] q = CellUtil.cloneQualifier(cell);
                    byte[] v = CellUtil.cloneValue(cell);
                    map.put(Bytes.toString(q), v);
                }
            } else {
                logger.warn("Table " + tableName + " does not exist.");
            }
        } catch (IOException e) {
            logger.error("getMapByKeyAndFamily " + tableName + " error.", e);
        } finally {
            closeTable(hTable);
        }

        return map;
    }

    /**
     * 获取一整行的值
     *
     * @param tableName
     * @param rowKey
     * @return
     * @author
     */
    public Map<String, byte[]> getRowMap(String tableName, String rowKey) {
        Table hTable = null;
        Map<String, byte[]> map = null;
        try {
            Get get = new Get(Bytes.toBytes(rowKey));
            hTable = getTable(tableName);
            if (getHBaseAdmin().tableExists(TableName.valueOf(tableName))) {
                map = new HashMap<String, byte[]>();
                Result result = hTable.get(get);
                for (Cell cell : result.rawCells()) {
                    byte[] q = CellUtil.cloneQualifier(cell);
                    byte[] v = CellUtil.cloneValue(cell);
                    map.put(Bytes.toString(q), v);
                }
            } else {
                logger.warn("Table " + tableName + " does not exist.");
            }
        } catch (IOException e) {
            logger.error("getRowMap " + tableName + " error", e);
        } finally {
            closeTable(hTable);
        }

        return map;
    }

    public void printResult(Result rs) {
        if (rs != null && !rs.isEmpty()) {
            for (Cell cell : rs.rawCells()) {
                System.out.println("\t列簇->" + new String(CellUtil.cloneFamily(cell)));
                System.out.println("\t列名->" + new String(CellUtil.cloneQualifier(cell)));
                System.out.println("\t列值->" + new String(CellUtil.cloneValue(cell)));
                System.out.println("------------------------------------------------------------------------------------");
            }
        }
    }

    /**
     * 插入单行单列簇单列修饰符数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param col
     * @param val
     */
    public void addOneRecord(String tableName, String rowKey, String family, String col, String val) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            Put p = new Put(Bytes.toBytes(rowKey));
            p.addColumn(Bytes.toBytes(family), Bytes.toBytes(col), Bytes.toBytes(val));
            if (p.isEmpty()) {
                logger.warn("*****数据插入异常，请确认数据完整性，稍候重试");
            } else {
                hTable.put(p);
            }
        } catch (IOException e) {
            logger.error("*****addOneRecord出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 批量插入数据
     *
     * @param tableName
     * @param dataList
     */
    public void addBatchRecords(String tableName, List<HbasePutsDTO> dataList) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            List<Put> puts = dataList.stream().filter(Objects::nonNull)
                    .filter(it -> StringUtils.isNotBlank(it.getRowKey()))
                    .filter(it -> StringUtils.isNotBlank(it.getFamily()))
                    .filter(it -> MapUtils.isNotEmpty(it.getDataMap()))
                    .map(it -> {
                        Put put = new Put(Bytes.toBytes(it.getRowKey()));
                        it.getDataMap().forEach((key, value) -> {
                            if (StringUtils.isNotBlank(key)) {
                                put.addColumn(Bytes.toBytes(it.getFamily()), Bytes.toBytes(key), Bytes.toBytes(value));
                            } else {
                                logger.warn("*****列值为空，请确认数据完整性，稍候重试");
                            }
                        });

                        return put;
                    }).collect(Collectors.toList());

            if (!CollectionUtils.isEmpty(puts)) {
                hTable.put(puts);
            } else {
                logger.warn("*****数据插入异常，请确认数据完整性，稍候重试");
            }
        } catch (IOException e) {
            logger.error("*****addOneRecord出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 批量插入数据
     *
     * @param tableName
     * @param dataList
     */
    public void addBatchRecord(String tableName, List<HbasePutDTO> dataList) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            List<Put> puts = dataList.stream().filter(Objects::nonNull)
                    .filter(it -> StringUtils.isNotBlank(it.getRowKey()))
                    .filter(it -> StringUtils.isNotBlank(it.getFamily()))
                    .filter(it -> StringUtils.isNotBlank(it.getCol()))
                    .filter(it -> StringUtils.isNotBlank(it.getValue()))
                    .map(it -> {
                        Put put = new Put(Bytes.toBytes(it.getRowKey()));
                        put.addColumn(Bytes.toBytes(it.getFamily()), Bytes.toBytes(it.getCol()), Bytes.toBytes(it.getValue()));

                        return put;
                    }).collect(Collectors.toList());

            if (!CollectionUtils.isEmpty(puts)) {
                hTable.put(puts);
            } else {
                logger.warn("*****数据插入异常，请确认数据完整性，稍候重试");
            }
        } catch (IOException e) {
            logger.error("*****addOneRecord出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 插入更新单行单列簇单列修饰符数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param col
     * @param val
     */
    @Deprecated
    public void addOrUpdateOneRecord(String tableName, String rowKey, String family, String col, String val) {
        Map<String, String> dataMap = new HashMap<String, String>();
        Result result = getRecordByRowKey(tableName, rowKey);
        if (!result.isEmpty()) {
            for (Cell cell : result.rawCells()) {
                byte[] q = CellUtil.cloneQualifier(cell);
                byte[] v = CellUtil.cloneValue(cell);
                dataMap.put(Bytes.toString(q), Bytes.toString(v));
            }
        }
        dataMap.put(col, val);

        addOneRecord(tableName, rowKey, family, dataMap);
    }


    /**
     * 插入单行单列簇多列修饰符数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param dataMap
     */
    public void addOneRecord(String tableName, String rowKey, String family, Map<String, String> dataMap) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            Put p = new Put(Bytes.toBytes(rowKey));
            dataMap.forEach((key, value) -> {
                if (StringUtils.isNotBlank(key)) {
                    p.addColumn(Bytes.toBytes(family), Bytes.toBytes(key), Bytes.toBytes(value));
                } else {
                    logger.warn("*****列值为空，请确认数据完整性，稍候重试");
                }
            });
            //当put对象没有成功插入数据时，此时调用hTable.put(p)方法会报错：java.lang.IllegalArgumentException:No columns to insert
            if (p.isEmpty()) {
                logger.warn("*****数据插入异常，请确认数据完整性，稍候重试");
            } else {
                hTable.put(p);
            }
        } catch (IOException e) {
            logger.error("*****addMoreRecord出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 插入更新单行单列簇多列修饰符数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param dataMap
     */
    @Deprecated
    public void addOrUpdateOneRecord(String tableName, String rowKey, String family, Map<String, String> dataMap) {
        Result result = getRecordByRowKey(tableName, rowKey);
        if (!result.isEmpty()) {
            Map<String, String> existMap = new HashMap<String, String>();
            for (Cell cell : result.rawCells()) {
                byte[] q = CellUtil.cloneQualifier(cell);
                byte[] v = CellUtil.cloneValue(cell);
                existMap.put(Bytes.toString(q), Bytes.toString(v));
            }
            existMap.putAll(dataMap);
            dataMap = existMap;
        }

        Table hTable = null;
        try {
            hTable = getTable(tableName);
            Put p = new Put(Bytes.toBytes(rowKey));
            dataMap.forEach((key, value) -> {
                if (StringUtils.isNotBlank(key)) {
                    p.addColumn(Bytes.toBytes(family), Bytes.toBytes(key), Bytes.toBytes(value));
                } else {
                    logger.warn("*****列值为空，请确认数据完整性，稍候重试");
                }
            });
            //当put对象没有成功插入数据时，此时调用hTable.put(p)方法会报错：java.lang.IllegalArgumentException:No columns to insert
            if (p.isEmpty()) {
                logger.warn("*****数据插入异常，请确认数据完整性，稍候重试");
            } else {
                hTable.put(p);
            }
        } catch (IOException e) {
            logger.error("*****addMoreRecord出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 删除指定行
     *
     * @param tableName
     * @param rowKey
     */
    public void deleteRow(String tableName, String rowKey) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            hTable.delete(new Delete(Bytes.toBytes(rowKey)));
        } catch (IOException e) {
            logger.error("*****deleteRow出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 删除指定行-列簇
     *
     * @param tableName
     * @param rowKey
     * @param family
     */
    public void deleteRow(String tableName, String rowKey, String family) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            hTable.delete(new Delete(Bytes.toBytes(rowKey)).addFamily(Bytes.toBytes(family)));
        } catch (IOException e) {
            logger.error("*****deleteRow出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }

    /**
     * 删除指定行-列簇-列
     *
     * @param tableName
     * @param rowKey
     * @param family
     */
    public void deleteRow(String tableName, String rowKey, String family, String qualifier, boolean deleteAll) {
        Table hTable = null;
        try {
            hTable = getTable(tableName);
            if (deleteAll) {
                hTable.delete(new Delete(Bytes.toBytes(rowKey)).addColumns(Bytes.toBytes(family), Bytes.toBytes(qualifier)));
            } else {
                hTable.delete(new Delete(Bytes.toBytes(rowKey)).addColumn(Bytes.toBytes(family), Bytes.toBytes(qualifier)));
            }
        } catch (IOException e) {
            logger.error("*****deleteRow出现错误:" + e.getMessage(), e);
        } finally {
            closeTable(hTable);
        }
    }
}